
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

import java.util
/**
 * @Time ： 2024/10/24
 * @Auth ： cheng
 * @IDE ：ide
 * @Motto：ABC(Always Be Coding)
 * */
object HelloSparkStream {


  def main(args: Array[String]): Unit = {

    /*  maven 依赖  之前 spark-core
        spark streaming   spark-streaming jar
        spark-streaming-kafka

       1) 创建 sparkconf 配置项 本地，应用程序名字
       2） 创建 sparkStreaming  加载conf （ 实时的数据流）  ssc
       3)  设置 编码  日志级别 error
       4）  设置 kafka topic ，key value 反序列化， groupid ，offset 提交
       5） 订阅 kafka的 消费者  topic ssc
       6） 消费topic 当中的数据


     */
    //创建 sparkconf 配置项 本地，应用程序名字
    val conf = new SparkConf().setMaster("local[*]").setAppName("helloStream")
    //创建 sparkStreaming  加载conf （ 实时的数据流）  ssc
    val ssc = new StreamingContext(conf,Seconds(2))

    // error
    ssc.sparkContext.setLogLevel("error")

    // kafka 配置
    val kfkParams = Map[String,Object](
      "bootstrap.servers" ->"123.56.187.176:9092",
      "key.deserializer"  -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "cheng",
      "enable.auto.commit" -> (false:java.lang.Boolean)
    )

    // topic
    val topicName = Array("stuInfo")
    //    val topicName = Array("test1")

    // kafkaUtils  导入一个依赖 spark-streaming-kafka
    val streamRdd = KafkaUtils.createDirectStream[String,String](
      ssc,
      PreferConsistent, // 位置策略
      Subscribe[String,String](topicName,kfkParams)  // 订阅topic的名字
    )


    //时间窗口函数
    // kafka 的 value
    //streamRdd kafka 返回的对象，值只是在value	RDD (key,value )
    // 1) 拿到kafka的数据value
    // val line = streamRdd.map(x => {x.value()} )v
    val line = streamRdd.map( _.value() ) // 空格，\t， 分隔符 是\t
    val res = line.flatMap(_.split(" ")).map((_,1)).reduceByKeyAndWindow(
      _+_,
      Seconds(4),
      Seconds(4)
    )


    //----	--- producer 的配置项
    // res Dstream --> RDD(Array)
    res.foreachRDD( x =>{
    println("------- 数据 结果 -----	---------------------")

    x.foreach(println(_))

    // 构建 producer 发送
  }
  )

//    val line =streamRdd.foreachRDD(
//      x=>{
//        if(x.isEmpty()){
//          val rs1 = x.map(_.value())
//        }
//      }
//    )

    // 窗口函数
    //line

    //  producer 配置项
    val property = new util.HashMap[String,Object]()
    property.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.47.129:9092")
    property.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    property.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")



    streamRdd.foreachRDD(
      x=>{
        // 不为空的时候 进行打印
        if(!x.isEmpty()){
          // 取 kafka当中的 value ，为实际的数据
          val line = x.map(_.value())
          // 对一行数据进行打印
          //          line.foreach(println)
          // hello java
          // 词频统计
          val result = line.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
          result.foreach(println)



          result.foreach(
            obj =>{
              // obj  元素

              val kfkProducer = new KafkaProducer[String,String](property)
              kfkProducer.send(   new ProducerRecord[String,String]    (  "16testnew",obj.toString()   ))
              kfkProducer.close()


            }
          )










          // 窗口函数
        }
      }
    )






    ssc.start()  //开启 ssc
    ssc.awaitTermination() //等待数据输入

  }

}

